package com.freez.spark.core.demo;


import com.freez.spark.tool.TupleUtils;
import org.apache.spark.Partitioner;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

import java.util.ArrayList;
import java.util.List;

/**
 * FREEDOM  2021 人生苦短，不妨一试
 * FileName: SumDemo.java
 * Author: zcs
 * Date: 2021年-12月-07日 周二 18:33
 * Description: TopN算法
 */
public class TopN {
    public static void main(String[] args) {
        String appName = "SparkDemo-TopN";
        String master = "local";
//        String master = "spark://slaveNode2:7077";
        //通过HDFS获取数据源
        String path = "./src/main/resources/data/core/demo/sparkTopN.txt";
        //选择输出人数
        int selectNum = 10;
        topN(appName, master, path, selectNum);
    }

    public static void topN(String appName, String master, String path, int selectNum) {
        SparkConf conf = new SparkConf();
        conf.setAppName(appName).setMaster(master);
        JavaSparkContext jsc = new JavaSparkContext(conf);
        JavaRDD<String> javaRDD = jsc.textFile(path);
        JavaPairRDD<String, Tuple2<Integer, String>> partRDD = javaRDD.mapToPair(s -> {
            String[] split = s.split(",");
            String name = split[0];
            String part = split[1];
            Integer score = Integer.parseInt(split[2]);
            Tuple2<Integer, String> value = TupleUtils.T2(score, name);
            return TupleUtils.T2(part, value);
        });
        /*JavaPairRDD<String, Integer> nameAndScoreRDD = javaRDD.mapToPair(s -> {
            String[] split = s.split(",");
            String name = split[0];
            Integer score = Integer.parseInt(split[2]);
            return TupleUtils.T2(name, score);
        });
        JavaPairRDD<String, Iterable<Integer>> scoreOfPart = nameAndScoreRDD.groupByKey();
        JavaPairRDD<String, List<Tuple2<Integer, String>>> groupPart = partRDD.groupByKey().mapValues(s -> {
            List<Tuple2<Integer, String>> list = Lists.newArrayList(s);
            //各科之间进行排名
            Collections.sort(list, new Comparator<Tuple2<Integer, String>>() {
                @Override
                public int compare(Tuple2<Integer, String> o1, Tuple2<Integer, String> o2) {
                    return o1._1 > o2._1 ? -1 : 1;
                }
            });
            return selectTopN(list, selectNum);
        });
        //输出各科的排名情况
        for (Tuple2<String, List<Tuple2<Integer, String>>> item : groupPart.collect()) {
            StringBuilder stringBuilder = new StringBuilder();
            for (Tuple2<Integer, String> a : item._2) {
                stringBuilder.append("," + a);
            }
            System.out.println("学科：" + item._1 + "，排名情况：" + stringBuilder.toString().replaceFirst(",", ""));
        }
        JavaPairRDD<String, Integer> tempScore = groupPart.flatMapToPair(s -> {
            List<Tuple2<Integer, String>> list = Lists.newArrayList(s._2);
            List<Tuple2<String, Integer>> resultList = new ArrayList<>();
            if (list.isEmpty()) {
                return null;
            }
            for (Tuple2<Integer, String> item : list) {
                resultList.add(TupleUtils.T2(item._2, item._1));
            }
            return resultList.iterator();
        });
        //按照姓名将三科分数进行连接，计算总分
        JavaPairRDD<String, Integer> sumScoreRDD = tempScore.leftOuterJoin(scoreOfPart).groupByKey().mapValues(s -> {
            List<Tuple2<Integer, Optional<Iterable<Integer>>>> list = Lists.newArrayList(s);
            int sumScore = 0;
            if (!list.isEmpty()) {
                Optional<Iterable<Integer>> optional = list.get(0)._2;
                if (optional.isPresent()) {
                    List<Integer> sumScoreList = Lists.newArrayList(optional.get());
                    for (Integer score : sumScoreList) {
                        sumScore = sumScore + score;
                    }
                }
            }
            return sumScore;
        });
        JavaPairRDD<Integer, Iterable<String>> resultRDD = sumScoreRDD.mapToPair(s -> s.swap()).groupByKey().sortByKey(false);
        //输出选择人数之间的总分数并排名（可能为总排名也可能不是总排名）
        resultRDD.foreach(s -> System.out.println(s));*/
//        partRDD.repartition(3);
//        partRDD.partitionBy(new myPartitioner()).foreach(s -> System.out.println(s));
    }

    /**
     * 选择各科前多少名的学生
     */
    public static List<Tuple2<Integer, String>> selectTopN(List<Tuple2<Integer, String>> list, int sortNum) {
        List<Tuple2<Integer, String>> resultList = new ArrayList<>();
        for (int i = 0; i < sortNum; i++) {
            resultList.add(list.get(i));
        }
        return resultList;
    }
}

class myPartitioner extends Partitioner {
    private int partitions;

    @Override
    public int numPartitions() {
        return 2;
    }

    @Override
    public int getPartition(Object key) {
        return 0;
    }
}